package com.example.hbase;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

import static org.apache.hadoop.hbase.HConstants.CLUSTER_DISTRIBUTED;
import static org.apache.hadoop.hbase.HConstants.ZOOKEEPER_QUORUM;

/**
 * @author Administrator
 */
@Component
public class HBaseComponent {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    static String zkAddress = "hb-2zez9857h1otyjl10-001.hbase.rds.aliyuncs.com:2181,hb-2zez9857h1otyjl10-002.hbase.rds.aliyuncs.com:2181,hb-2zez9857h1otyjl10-003.hbase.rds.aliyuncs.com:2181";

    static Configuration config = HBaseConfiguration.create();

    static String tableName = "leniao-realvar1";

    public static final ConcurrentHashMap<String, String> CONCURRENT_HASH_MAP = new ConcurrentHashMap<>();

    static {
        config.set(CLUSTER_DISTRIBUTED, "true");
        config.set(ZOOKEEPER_QUORUM, zkAddress);
    }

    String startTime = "2018-12-30 11:12";
    String pattern = "yyyy-MM-dd HH:mm";

    @Async("taskExecutor")
    @Scheduled(cron = "00 */1 * * * ?")
    public void scheduler() {
        String time = CONCURRENT_HASH_MAP.get("time");
        DateTimeFormatter dateTimeFormatter = DateTimeFormat.forPattern(this.pattern);
        if (StringUtils.isBlank(time)) {
            DateTime dateTime = DateTime.parse(startTime, dateTimeFormatter).minusMinutes(1);
            time = dateTime.toString(this.pattern);
        } else {
            DateTime dateTime = DateTime.parse(time, dateTimeFormatter).minusMinutes(1);
            time = dateTime.toString(this.pattern);
        }
        CONCURRENT_HASH_MAP.put("time", time);
        logger.info("\n删除时间点：{}", time);
        this.execute(time);
    }

    private void execute(String time) {
        ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, new SubstringComparator(time));
        Scan scan = new Scan();
        scan.setFilter(valueFilter);
        try (Connection connection = ConnectionFactory.createConnection(config);
             Table table = connection.getTable(TableName.valueOf(tableName));
             ResultScanner scanner = table.getScanner(scan)) {
            List<Delete> deleteList = new ArrayList<>();
            for (Result res : scanner) {
                deleteList.add(new Delete(res.getRow()));
            }
            table.delete(deleteList);
            logger.info("\n删除数据受影响行数： {}", deleteList.size());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}